
#include "../iface/debug.h"
#include "../aprtools/APRAutoLock.h"
#include "ForwardServer.h"

/** Start handling clients to the forwarding pipe.
 * 
 * @param threadid The integer thread ID number.
 * 
 * @param localmp The memory pool local to this thread. */
void ForwardServer::handleClientsOnThread(int threadid, apr_pool_t *localmp)
{
   pipearray[threadid].sessionstart->lock();
   while (!sessionstop)
   {
      pipearray[threadid].sessionstart->wait();
      if (sessionstop) break;
      if (fchannel)
      {
         pipearray[threadid].waiting->lock();
         char *buffer = (char *)malloc(sizeof(char)*FORWARD_STREAMBUFFER);
         int err = 0;
         int timeout = 1;
         while ((pipearray[threadid].handler_to_send) && (pipearray[threadid].recv_to_handler) &&
                (!(pipearray[threadid].handler_to_send->isClosed())) && 
                (!(pipearray[threadid].recv_to_handler->isClosed())) && (!err))
         {
            if (!(pipearray[threadid].recv_to_handler->hasData()))
            {
               // always returns locked
               int acquire = pipearray[threadid].waiting->timedWait(timeout);
               if (acquire < 0) // error condition, probably timed out
               {
                  timeout <<= 1;
                  timeout = (timeout > SSH_TIMEOUT_EAGAIN) ? SSH_TIMEOUT_EAGAIN : timeout;
               }
               int read_from_fchan = 0;
               {
                  APRAutoLock autolock(sshmutex);
                  read_from_fchan = libssh2_channel_read(fchannel, buffer, FORWARD_STREAMBUFFER);
               }
               if (read_from_fchan == LIBSSH2_ERROR_EAGAIN)
               {
                  read_from_fchan = 0;
               } else if (read_from_fchan < 0)
               {
                  switch (read_from_fchan)
                  {
                     case LIBSSH2_ERROR_SOCKET_RECV:
                        errprint("error receiving on socket");
                        break;
                     default: break;
                  }
                  errprint("could not read from channel (%d)", read_from_fchan);
                  err |= read_from_fchan;
                  break;
               }
               pipearray[threadid].handler_to_send->writeToStream(buffer, read_from_fchan);
            }
            if ((pipearray[threadid].handler_to_send) && (pipearray[threadid].recv_to_handler))
            {
               while ((pipearray[threadid].handler_to_send->packedSpaceRemaining() >= FORWARD_STREAMBUFFER) && 
                      (pipearray[threadid].recv_to_handler->hasData()))
               {
                  apr_size_t dataquant = pipearray[threadid].recv_to_handler->popFromStream(buffer, FORWARD_STREAMBUFFER);
                  int written_to_fchan = 0;
                  int ierr;
                  ierr = 1;
                  while ((ierr >= 0) && (written_to_fchan < dataquant))
                  {
                     {
                        APRAutoLock autolock(sshmutex);
                        ierr = libssh2_channel_write(fchannel, &(buffer[written_to_fchan]), dataquant - written_to_fchan);
                     }
                     if (ierr == LIBSSH2_ERROR_EAGAIN)
                     {
                        ierr = 0;
                     } else if (ierr < 0)
                     {
                        switch (ierr)
                        {
                           case LIBSSH2_ERROR_SOCKET_SEND:
                              errprint("error sending on socket");
                              break;
                           default: break;
                        }
                        errprint("could not write to channel (%d)", ierr);
                        err |= ierr;
                        break;
                     }
                     written_to_fchan += ierr;
                  }
               }
            }
         }
         if (pipearray[threadid].recv_to_handler)
            pipearray[threadid].recv_to_handler->writeEOS();
         if (pipearray[threadid].handler_to_send)
            pipearray[threadid].handler_to_send->writeEOS();
         free(buffer);
         pipearray[threadid].waiting->unlock();
      }
   }
   pipearray[threadid].sessionstart->unlock();
}

/** Start handling clients to the forwarding pipe (the client handler
 * threads run this).
 * 
 * @param tthread The APR pointer to this thread grabbed.
 * 
 * @param args The void pointer given to the thread. This is actually
 * a pointer to an array of two void pointers, which are the this
 * pointer and a pointer to the thread ID respectively.
 * 
 * @return Return is NULL. */
void *ForwardServer::handleClientsOnThreadStatic(apr_thread_t *tthread, void *args)
{
   apr_status_t status;
   ForwardServer *kthis = *((ForwardServer **)(((void **)(args))[0]));
   int i = *((int *)(((void **)(args))[1]));
   apr_pool_t *upperpool, *mp;
   upperpool = apr_thread_pool_get(tthread);
   status = apr_pool_create(&mp, upperpool);
   CheckAPRError(status);

   kthis->handleClientsOnThread(i, mp);

   apr_pool_clear(mp);
   apr_pool_destroy(mp);
   ERR_remove_state(0);
   apr_thread_exit(tthread, APR_SUCCESS);
   return NULL;
}

/** Sets the forwarding channel that this forwarding server forwards
 * data to and from. Should only be called once, prior to
 * initialisation.  The channel should be valid while this class is in
 * scope.
 * 
 * @param ifchannel The LIBSSH2 format channel that the forwarding
 * server must use. */
void ForwardServer::setForwardingChannel(LIBSSH2_CHANNEL *ifchannel)
{
   fchannel = ifchannel;
}
